package client;


import com.google.common.base.Strings;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.client.*;
import org.apache.ignite.configuration.ClientConfiguration;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import other.IgniteJobConfig;
import other.IgniteUtil;

import java.util.*;

public class KVTest {

    private static final String CACHE_NAME = "BIGDATA_TEST";
    private static final Set<String> COMMANDSET = new HashSet<>();
    private static volatile Boolean run = true;
    private static final String CACHE_TRANSCATION_NAME = "BIGDATA_TEST_1";

    static {
        COMMANDSET.add("GET");
        COMMANDSET.add("GETALL");
        COMMANDSET.add("PUT");
        COMMANDSET.add("PUTALL");
        COMMANDSET.add("ATOMIC");
    }

    public static void main(String[] args) {

        IgniteJobConfig igniteJobConfig = IgniteUtil.parseConfig();
        ClientConfiguration clientConfiguration = IgniteUtil.init(igniteJobConfig.getUri());

        Scanner scanner = new Scanner(System.in);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                Thread.sleep(200);
                System.out.println("Shouting down ...");
                run = false;
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }));

        try (IgniteClient igniteClient = Ignition.startClient(clientConfiguration)) {
            System.out.println("请按照下方命令输入：\n" +
                    "GET key\n" +
                    "PUT key value\n" +
                    "PUTALL key1 value key2 value\n" +
                    "GETALL key1 key2\n" +
                    "ATOMIC Y/N  Y为事务异常时测试 N为事务无异常时测试\n"
//                    "ATOMIC 所用的cache与 GET PUT等命令所用的cache不一致"
            );
            // 创建cache 操作对象
            ClientCache<Object, Object> cache = igniteClient.getOrCreateCache(CACHE_NAME);
            while (run) {
                try {
                    System.out.println("请输入：");
                    String command = scanner.nextLine();
                    if (!Strings.isNullOrEmpty(command)) {
                        switchCommand(command, igniteClient, cache);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static void switchCommand(String command, IgniteClient igniteClient, ClientCache<Object, Object> cache) {


        String trim = command.trim();
        String commandType = getCommandType(trim);
        switch (commandType) {
            case "GET":
                Set<Object> keys = getKeys(trim);
                Object o = cache.get(keys.toArray()[0]);
                System.out.println(keys.toArray()[0] + " " + o);
                break;
            case "GETALL":
                Set<Object> keys1 = getKeys(trim);
                Map<Object, Object> all = cache.getAll(keys1);
                all.forEach((key, value) -> System.out.println(key + " " + value));
                break;
            case "PUT":
                Map<Object, Object> setKvs = getSetKvs(trim);
                Map.Entry<Object, Object> a = (Map.Entry<Object, Object>) setKvs.entrySet().toArray()[0];
                cache.put(a.getKey(), a.getValue());
                System.out.println("操作成功");
                break;
            case "PUTALL":
                Map<Object, Object> setKvs1 = getSetKvs(trim);
                cache.putAll(setKvs1);
                System.out.println("操作成功");
                break;
            case "ATOMIC":
                atomic(igniteClient, command);
                break;
            default:
                throw new IllegalStateException("Unexpected value: " + commandType);
        }

    }

    private static String getCommandType(String trim) {
        int commandIndex = trim.indexOf(" ");
        String substring = commandIndex == -1 ? trim : trim.substring(0, commandIndex).toUpperCase();
        if (COMMANDSET.contains(substring)) {
            return substring;
        } else {
            throw new RuntimeException("输入格式错误");
        }
    }

    private static Map<Object, Object> getSetKvs(String trim) {
        String[] s1 = trim.split(" ");
        int i = 0;
        Map<Object, Object> maps = new HashMap<>();
        String key = null;
        for (String s : s1) {
            if (!Strings.isNullOrEmpty(s) && !COMMANDSET.contains(s.toUpperCase())) {
                if (i % 2 == 0) {
                    key = s;
                } else {
                    maps.put(key, s);
                }
                i++;
            }
        }
        if (i % 2 != 0 || i == 0) {
            throw new RuntimeException("输入格式错误");
        }
        return maps;
    }

    private static Set<Object> getKeys(String trim) {
        String[] s = trim.split(" ");
        Set<Object> keys = new HashSet<>();
        for (String s1 : s) {
            if (!Strings.isNullOrEmpty(s1) && !COMMANDSET.contains(s1)) {
                keys.add(s1);
            }
        }
        return keys;
    }

    private static void atomic(IgniteClient client, String command) {
        //client cache 配置
        ClientCacheConfiguration clientCacheConfiguration = new ClientCacheConfiguration();
        clientCacheConfiguration.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
        clientCacheConfiguration.setName(CACHE_TRANSCATION_NAME);
        clientCacheConfiguration.setBackups(1);
        //
        ClientCache<Object, Object> cache = client.getOrCreateCache(clientCacheConfiguration);
        ClientTransactions transactions = client.transactions();
        System.out.println("即将进行原子操作，先设置初始值(key1, value1),(key2, value2)");
        cache.put("key1", "value1");
        cache.put("key2", "value2");
        System.out.println("进入事务前初始值");
        Object key = cache.get("key1");
        System.out.println("key1 " + key);

        Object key2 = cache.get("key2");
        System.out.println("key2 " + key2);
        try (ClientTransaction tx = transactions.txStart(TransactionConcurrency.OPTIMISTIC,
                TransactionIsolation.SERIALIZABLE)) {
            Object key1 = cache.get("key1");
            if ("value1".equals(key1)) {
                cache.put("key1", "value1111");
                System.out.println("事务中，修改key1 值为value1111");
            }
            if (command.toUpperCase().contains("Y")) {
                throwException();
            }
            cache.put("key2", "value2222");
            System.out.println("事务中，修改key2 值为value2222");
            tx.commit();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("事务后，当前值");

        Object key11 = cache.get("key1");
        System.out.println("key1 " + key11);

        Object key22 = cache.get("key2");
        System.out.println("key2 " + key22);
    }


    private static void throwException() {
        throw new RuntimeException("事务异常测试");
    }

}
